package net.bwie.realtime.jtp.dwd.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class LogSplitProcessFunction extends ProcessFunction<String, String> {
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;

    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public LogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        // 1.解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        // 2.错误数据
        if (null!=jsonObject.getString("err")){
            ctx.output(errorTag,value);
            jsonObject.remove("err"); // 删除err字段
        }
        // 3.启动日志
        if (null!=jsonObject.get("start")){
            //侧边流输出
            ctx.output(startTag,jsonObject.toJSONString());
        }

        else {
            // 4.曝光日志和点击日志
            JSONArray displays = jsonObject.getJSONArray("display");
            JSONArray actions = jsonObject.getJSONArray("actions");
            jsonObject.remove("display");
            jsonObject.remove("actions");
            out.collect(jsonObject.toJSONString());
            if (null != displays && !displays.isEmpty()) {
                for (int i = 0; i < displays.size(); i++) {
                    JSONObject display = displays.getJSONObject(i);
                    display.put("display", display);
                    ctx.output(displayTag, display.toJSONString());
                }
            }

            jsonObject.remove("actions");
            if (null != actions && !actions.isEmpty()) {
                for (int i = 0; i < actions.size(); i++) {
                    JSONObject action = actions.getJSONObject(i);
                    action.put("action", action);
                    ctx.output(actionTag, action.toJSONString());
                }
            }
        }
    }
}
